package com.dzx;

import com.kumkee.userAgent.UserAgent;
import com.kumkee.userAgent.UserAgentParser;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author DuanZhaoXu
 * @ClassName:
 * @Description: 使用hadoop mapreduce 统计页面的各个浏览器访问次数
 * @date 2018年12月05日 13:38:17
 */

public class LogApp {
    public static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

        UserAgentParser userAgentParser = null;

        /**
         * setup 在 map 之前只会执行一次
         *
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            userAgentParser = new UserAgentParser();
        }

        /**
         * cleanup 在 map之后只会执行一次
         *
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            userAgentParser = null;
        }

        LongWritable one = new LongWritable(1);

        /**
         * map 读取输入的文件
         *
         * @param key
         * @param value
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //接收到的文件的每一行日志记录
            String log = value.toString();
            String userAgentString = log.substring(getCharacterPosition(log, "\"", 7) + 1);
            UserAgent userAgent = userAgentParser.parse(userAgentString);
            String platform = userAgent.getPlatform();
            context.write(new Text(platform), one);
        }

        /**
         * 获取指定字符串中指定标识的字符串出现的索引位置
         *
         * @param value
         * @param operator
         * @param index
         * @return
         */
        private int getCharacterPosition(String value, String operator, int index) {
            Matcher matcher = Pattern.compile(operator).matcher(value);
            int mIdx = 0;
            while (matcher.find()) {
                mIdx++;
                if (mIdx == index) {
                    break;
                }
            }
            return matcher.start();
        }
    }


    public static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
        /**
         * reduce
         *
         * @param key
         * @param values
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {

            long sum = 0;
            for (LongWritable longWritable : values) {
                sum += longWritable.get();
            }
            context.write(key, new LongWritable(sum));

        }
    }

    /**
     * 定义 Driver ：封装了MapReduce作业的所以信息
     * <p>
     * String[] args 参数数组  ： args[0]  输入文件路径  args[1]输出文件
     * * @param args
     */
    public static void main(String[] args) throws Exception {
        //创建
        Configuration configuration = new Configuration();
        //mapReduce 再次执行同样的任务，如果同名输出文件已存在会报错，所以首先如果文件已存在，先进行删除操作
        FileSystem fileSystem = FileSystem.newInstance(configuration);
        Path outPath = new Path(args[1]);
        if (fileSystem.exists(outPath)) {
            System.out.println("删除已经存在的 " + args[1] + "文件");
            fileSystem.delete(outPath, true);
        }
        //创建 job
        Job job = Job.getInstance(configuration, "logcount");
        // 设置job的处理类
        job.setJarByClass(LogApp.class);
        //设置作业处理的输入路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        //设置map的相关参数
        job.setMapperClass(MyMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        //设置combiner，相当于本地先进行了一次reduce，当map的结果传输到reduce的时候减少了网络传输量
        job.setCombinerClass(MyReducer.class);

        //设置reduce的相关参数
        job.setReducerClass(MyReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        //设置作业处理的输出路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
